package net.bw.realtime.tmall.dwd.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bw.realtime.tmall.common.utils.JdbcUtil;
import net.bw.realtime.tmall.common.utils.KafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

/*
 * @ Author：liuyawei
 * @ Date：2025-05-26
 */
public class sinkClickhouseFunction {

    public static void saveTmallLog(DataStream<String> stream){

        // 1.转换格式
        SingleOutputStreamOperator<String> jsonStream = stream.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        // 1解析JSON

                        JSONObject object = JSON.parseObject(value);
                        JSONObject productJSONObject = object.getJSONObject("actionDetail");
                        JSONObject deviceJSONObject = object.getJSONObject("deviceInfo");
                        JSONObject locationJSONObject = object.getJSONObject("location");
                        JSONObject sessionJSONObject = object.getJSONObject("session");

                        // 2获取订单数据
                        JSONObject jsonObject = new JSONObject();
                        jsonObject.put("orderId", object.getString("orderId"));
                        jsonObject.put("userId", object.getString("userId"));
                        jsonObject.put("actionType", object.getString("actionType"));
                        jsonObject.put("skuId", productJSONObject.getOrDefault("skuId", ""));
                        jsonObject.put("category", productJSONObject.getOrDefault("category", ""));
                        jsonObject.put("price", productJSONObject.getLong("price"));
                        jsonObject.put("brand", productJSONObject.getOrDefault("brand", ""));

                        // 3获取设备数据
                        jsonObject.put("deviceId", deviceJSONObject.getOrDefault("deviceId", ""));
                        jsonObject.put("deviceType", deviceJSONObject.getOrDefault("deviceType", ""));
                        jsonObject.put("os", deviceJSONObject.getOrDefault("os", ""));

                        // 4获取位置数据
                        jsonObject.put("province", locationJSONObject.getOrDefault("province", ""));
                        jsonObject.put("city", locationJSONObject.getOrDefault("city", ""));

                        // 5获取会话数据
                        jsonObject.put("sessionId", sessionJSONObject.getOrDefault("sessionId", ""));

                        // 6时间戳
                        jsonObject.put("ts", object.getLong("timestamp"));

                        // 7返回数据
                        return jsonObject.toJSONString();
                    }
                }
        );

        // 2.插入数据
        String[] columns = new String[]{
                "orderId", "userId", "actionType", "skuId", "category", "price", "brand",
                "deviceId", "deviceType", "os", "province", "city", "sessionId", "ts"
        };

        // 3.构造SQL
        String upsertSql = "INSERT INTO jtp_tmall_log.dwd_tmall_log(\n" +
                "    orderId,userId, actionType, skuId, category, price, brand, \n" +
                "    deviceId, deviceType, os, province, city, sessionId, ts\n" +
                ") VALUES (\n" +
                "    ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?\n" +
                ")" ;

        // 4.执行
        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream,columns, upsertSql);

    }

}
